package com.bw.app.dwd1;

import com.bw.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.Duration;


/**
 * @projectName: Flinksxe
 * @fileName: DwdTradeOrderDetail
 * @packageName: com.bw.app.dwd
 * @date: 2025/9/16 9:00
 * @author: ymw
 */
public class DwdTradeOrderDetail {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        //设置ttl状态 生产环境设置为最大乱序程度
        tenv.getConfig().setIdleStateRetention(Duration.ofSeconds(5L));
        tenv.executeSql("create table dwd_order_pre(\n" +
                "           id String,\n" +
                "           order_id String,\n" +
                "           sku_id String,\n" +
                "           sku_name String,\n" +
                "           order_price String,\n" +
                "           sku_num String,\n" +
                "           create_time String,\n" +
                "           source_type_id String,\n" +
                "           source_type_name String,\n" +
                "           `source_id` string,\n" +
                "           split_total_amount String,\n" +
                "           split_activity_amount String,\n" +
                "           split_coupon_amount String,\n" +
                "           consignee String,\n" +
                "           consignee_tel String,\n" +
                "           total_amount String,\n" +
                "           order_status String,\n" +
                "           user_id String,\n" +
                "           payment_way String,\n" +
                "           delivery_address String,\n" +
                "           order_comment String,\n" +
                "           out_trade_no String,\n" +
                "           trade_body String,\n" +
                "           oprate_time String,\n" +
                "           expire_time String,\n" +
                "           process_status String,\n" +
                "           tracking_no String,\n" +
                "           parent_order_id String,\n" +
                "           province_id String,\n" +
                "           activity_reduce_amount String,\n" +
                "           coupon_reduce_amount String,\n" +
                "           original_total_amount String,\n" +
                "           feight_fee String,\n" +
                "           feight_fee_reduce String,\n" +
                "           refundable_time String,\n" +
                "           order_detail_activity_id String,\n" +
                "           activity_id String,\n" +
                "           activity_rule_id String,\n" +
                "           order_detail_coupon_id String,\n" +
                "           coupon_id String,\n" +
                "           coupon_use_id String,\n" +
                "           `type` String,\n" +
                "           `old` Map<String,String>,\n" +
                "           row_op_ts  TIMESTAMP_LTZ(3)" +
                "  )" + MyKafkaUtil.getKafkaDDL("dwd_trade_order_pre_process","order_detail_2212A"));
        Table filtertable = tenv.sqlQuery("select\n" +
                "    id,\n" +
                "    order_id,\n" +
                "    user_id,\n" +
                "    sku_id,\n" +
                "    sku_name,\n" +
                "    sku_num,\n" +
                "    order_price,\n" +
                "    province_id,\n" +
                "    activity_id,\n" +
                "    activity_rule_id,\n" +
                "    coupon_id,\n" +
                "    create_time,\n" +
                "    source_id,\n" +
                "    source_type_id,\n" +
                "    source_type_name,\n" +
                "    split_activity_amount,\n" +
                "    split_coupon_amount,\n" +
                "    split_total_amount,\n" +
                "    row_op_ts\n" +
                "from dwd_order_pre\n" +
                "where `type` = 'insert' ");
        tenv.createTemporaryView("filtered_table",filtertable);
        tenv.toAppendStream(filtertable, Row.class).print("filtered_table =====>");
        tenv.executeSql("create table trade_order_detail(\n" +
                "    id String,\n" +
                "    order_id String,\n" +
                "    user_id String,\n" +
                "    sku_id String,\n" +
                "    sku_name String,\n" +
                "    sku_num String,\n" +
                "    order_price String,\n" +
                "    province_id String,\n" +
                "    activity_id String,\n" +
                "    activity_rule_id String,\n" +
                "    coupon_id String,\n" +
                "    create_time String,\n" +
                "    source_id String,\n" +
                "    source_type_id String,\n" +
                "    source_type_name String,\n" +
                "    split_activity_amount String,\n" +
                "    split_coupon_amount String,\n" +
                "    split_total_amount String,\n" +
                "    row_op_ts TIMESTAMP_LTZ(3)\n" +
                ")"+MyKafkaUtil.getKafkaSinkDDL("dwd_trade_order_detail"));
        tenv.executeSql("insert into trade_order_detail select * from filtered_table");
        env.execute("DwdTradeOrderDetail");
    }
}
